package com.elinshaw;

import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;

import java.util.Map;
import java.util.Random;

/**
 * spout数据源实现
 *
 * @author Administrator
 */
public class NumberSpout extends BaseRichSpout {

    /**
     * 创建数据源的发射器将tuple发射给下游
     */
    private SpoutOutputCollector collector;


    /**
     * Spout组件初始化被调用
     * 只会被调用一次
     * spout加载时打开一些资源
     *
     * @param map
     * @param topologyContext
     * @param spoutOutputCollector
     */
    @Override
    public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {
        System.err.println("open");
        this.collector = spoutOutputCollector;
    }

    /**
     * 核心方法
     * 产生数据源storm会不断调用也就是方法执行完后会马上重置并再次执行直到拓扑终止
     * nextTuple()方法是Spout实现的核心。
     * 也就是主要执行方法用于输出信息通过collector.emit方法发射
     */
    @Override
    public void nextTuple() {
        System.err.println("nextTuple");
        int number = new Random().nextInt(100);
        // 把数据封装到Tuple的Value对象中
        // 参数是可变参数
        Values value = new Values(number);
        //通过发射器把tuple发射给下游
        collector.emit(value);
    }

    /**
     * 声明Tuple的字段名，有顺序的
     * 用于声明数据格式即输出的一个Tuple中包含几个字段
     *
     * @param outputFieldsDeclarer
     */
    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        //声明tuple的key字段
        //tuple的key和value需要一一对应且数量保持一致
        outputFieldsDeclarer.declare(new Fields("number"));
    }


}
